package com.cx.utils;

import com.esotericsoftware.kryo.Kryo;
import com.esotericsoftware.kryo.io.Input;
import com.esotericsoftware.kryo.io.Output;
import com.esotericsoftware.kryo.util.DefaultInstantiatorStrategy;
import io.lettuce.core.protocol.LettuceCharsets;
import org.apache.commons.lang3.ArrayUtils;
import org.objenesis.strategy.StdInstantiatorStrategy;
import org.springframework.util.CollectionUtils;

import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
import java.util.List;

public class KryoUtil {

    //每个线程的 Kryo 实例
    private static final ThreadLocal<Kryo> kryoLocal = new ThreadLocal<Kryo>() {
        @Override
        protected Kryo initialValue() {
            Kryo kryo = new Kryo();

            /**
             * 不要轻易改变这里的配置！更改之后，序列化的格式就会发生变化，
             * 上线的同时就必须清除 Redis 里的所有缓存，
             * 否则那些缓存再回来反序列化的时候，就会报错
             */
            //支持对象循环引用（否则会栈溢出）
            kryo.setReferences(true); //默认值就是 true，添加此行的目的是为了提醒维护者，不要改变这个配置

            //不强制要求注册类（注册行为无法保证多个 JVM 内同一个类的注册编号相同；而且业务系统中大量的 Class 也难以一一注册）
            kryo.setRegistrationRequired(false); //默认值就是 false，添加此行的目的是为了提醒维护者，不要改变这个配置

            //Fix the NPE bug when deserializing Collections.
            ((DefaultInstantiatorStrategy) kryo.getInstantiatorStrategy())
                    .setFallbackInstantiatorStrategy(new StdInstantiatorStrategy());

            return kryo;
        }
    };

    /**
     * 获得当前线程的 Kryo 实例
     *
     * @return 当前线程的 Kryo 实例
     */
    public static Kryo getInstance() {

        return kryoLocal.get();
    }

    //-----------------------------------------------
    //          只序列化/反序列化对象
    //          序列化的结果里，不包含类型的信息
    //-----------------------------------------------

    /**
     * 将对象序列化为字节数组
     *
     * @param object 任意对象
     * @param <T> 对象的类型
     * @return 序列化后的字节数组
     */
    public static <T> byte[] writeObjectToByteArray(T object) {
        if (object == null) {
            throw new RuntimeException("序列化对象(" + object + ")!");
        }

        ByteArrayOutputStream byteArrayOutputStream = new ByteArrayOutputStream();
        Output output = new Output(byteArrayOutputStream);

        Kryo kryo = getInstance();
        kryo.writeObject(output, object);
        output.flush();

        try {
            return byteArrayOutputStream.toByteArray();
        } finally {
            CacheConsts.close(output);
            CacheConsts.close(byteArrayOutputStream);
        }
    }

    /**
     * 将对象序列化为 String
     *
     * @param object 任意对象
     * @param <T> 对象的类型
     * @return 序列化后的字符串
     */
    public static <T> String serialize(T object) {

        return new String(writeObjectToByteArray(object), LettuceCharsets.UTF8);
    }

    /**
     * 将字节数组反序列化为原对象
     *
     * @param byteArray writeToByteArray 方法序列化后的字节数组
     * @param clazz     原对象的 Class
     * @param <T>       原对象的类型
     * @return 原对象
     */
    @SuppressWarnings("unchecked")
    public static <T> T readObjectFromByteArray(byte[] byteArray, Class<T> clazz) {
        if (ArrayUtils.isEmpty(byteArray)) {
            throw new RuntimeException("反序列化对象发生异常,byte序列为空!");
        }

        ByteArrayInputStream byteArrayInputStream = new ByteArrayInputStream(byteArray);
        Input input = new Input(byteArrayInputStream);

        Kryo kryo = getInstance();
        try {
            return kryo.readObject(input, clazz);
        } finally {
            CacheConsts.close(input);
            CacheConsts.close(byteArrayInputStream);
        }
    }

    /**
     * 将 String 反序列化为原对象
     * 利用了 Base64 编码
     *
     * @param data   writeToString 方法序列化后的字符串
     * @param clazz 原对象的 Class
     * @param <T>   原对象的类型
     * @return 原对象
     */
    public static <T> T deserialize(String data, Class<T> clazz) {

        return readObjectFromByteArray(data.getBytes(LettuceCharsets.UTF8), clazz);
    }

    //-----------------------------------------------
    //          序列化/反序列化对象，及类型信息
    //          序列化的结果里，包含类型的信息
    //          反序列化时不再需要提供类型
    //-----------------------------------------------

    /**
     * 将对象【及类型】序列化为字节数组
     *
     * @param objList 任意对象
     * @param <T> 对象的类型
     * @return 序列化后的字节数组
     */
    public static <T> byte[] writeToByteArray(List<T> objList) {
        if (CollectionUtils.isEmpty(objList)) {
            throw new RuntimeException("序列化对象列表(" + objList + ")参数异常!");
        }

        ByteArrayOutputStream byteArrayOutputStream = new ByteArrayOutputStream();
        Output output = new Output(byteArrayOutputStream);

        Kryo kryo = getInstance();
        kryo.writeClassAndObject(output, objList);
        output.flush();

        try {
            return byteArrayOutputStream.toByteArray();
        } finally {
            CacheConsts.close(output);
            CacheConsts.close(byteArrayOutputStream);
        }
    }

    /**
     * 将对象【及类型】序列化为 String
     *
     * @param objList 任意对象
     * @param <T> 对象的类型
     * @return 序列化后的字符串
     */
    public static <T> String serializeList(List<T> objList) {

        return new String(writeToByteArray(objList), LettuceCharsets.UTF8);
    }

    /**
     * 将字节数组反序列化为原对象
     *
     * @param byteArray writeToByteArray 方法序列化后的字节数组
     * @param <T>       原对象的类型
     * @return 原对象
     */
    @SuppressWarnings("unchecked")
    public static <T> List<T> readFromByteArray(byte[] byteArray, Class<T> clazz) {
        if (ArrayUtils.isEmpty(byteArray)) {
            throw new RuntimeException("反序列化对象发生异常,byte序列为空!");
        }

        ByteArrayInputStream byteArrayInputStream = new ByteArrayInputStream(byteArray);
        Input input = new Input(byteArrayInputStream);

        Kryo kryo = getInstance();
        try {
            return (List<T>) kryo.readClassAndObject(input);
        } finally {
            CacheConsts.close(input);
            CacheConsts.close(byteArrayInputStream);
        }
    }

    /**
     * 将 String 反序列化为原对象
     *
     * @param data writeToString 方法序列化后的字符串
     * @param <T> 原对象的类型
     * @return 原对象
     */
    public static <T> List<T> deserializeList(String data, Class<T> clazz) {
        return readFromByteArray(data.getBytes(LettuceCharsets.UTF8), clazz);
    }
}